//! 集群内部通信协议模块
//! 
//! 实现Redis集群内部通信协议，包括：
//! - 协议版本管理和协商
//! - 消息序列化/反序列化
//! - 可靠性保证机制
//! - 消息优先级处理
//! - 协议兼容性处理

use crate::cluster::traits::{
    ClusterProtocol, ProtocolMessage as ProtocolMessageTrait, ProtocolStats,
    ReliabilityLevel, MessagePriority
};
use crate::cluster::{ClusterError, ClusterResult};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::time::interval;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

/// 集群协议版本
pub const CLUSTER_PROTOCOL_VERSION: u16 = 1;
pub const MIN_SUPPORTED_VERSION: u16 = 1;
pub const MAX_SUPPORTED_VERSION: u16 = 1;

/// 集群协议消息类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ClusterProtocolMessage {
    /// 版本协商消息
    VersionNegotiation {
        sender_id: String,
        proposed_version: u16,
        supported_versions: Vec<u16>,
        timestamp: u64,
    },
    
    /// 版本协商响应
    VersionNegotiationResponse {
        sender_id: String,
        target_id: String,
        agreed_version: u16,
        timestamp: u64,
    },
    
    /// 集群状态同步请求
    StateSync {
        sender_id: String,
        target_id: Option<String>,
        state_version: u64,
        partial_update: bool,
        payload: Vec<u8>,
        reliability: ReliabilityLevel,
        priority: MessagePriority,
    },
    
    /// 集群状态同步响应
    StateSyncResponse {
        sender_id: String,
        target_id: String,
        state_version: u64,
        success: bool,
        error_message: Option<String>,
        payload: Vec<u8>,
    },
    
    /// 槽位迁移协议消息
    SlotMigration {
        sender_id: String,
        target_id: String,
        slot: u16,
        migration_type: SlotMigrationType,
        data: Vec<u8>,
        sequence_number: u64,
        reliability: ReliabilityLevel,
        priority: MessagePriority,
    },
    
    /// 故障检测消息
    FailureDetection {
        sender_id: String,
        target_id: Option<String>,
        suspected_node: String,
        failure_type: FailureType,
        evidence: Vec<u8>,
        timestamp: u64,
        priority: MessagePriority,
    },
    
    /// 选举消息
    Election {
        sender_id: String,
        target_id: Option<String>,
        election_type: ElectionType,
        candidate_id: String,
        epoch: u64,
        vote_request: bool,
        payload: Vec<u8>,
        priority: MessagePriority,
    },
    
    /// 心跳消息
    Heartbeat {
        sender_id: String,
        target_id: Option<String>,
        timestamp: u64,
        load_info: LoadInfo,
        priority: MessagePriority,
    },
    
    /// 应答消息 (用于可靠性保证)
    Acknowledgment {
        sender_id: String,
        target_id: String,
        ack_message_id: String,
        success: bool,
        error_code: Option<u32>,
        timestamp: u64,
    },
    
    /// 自定义协议消息
    Custom {
        sender_id: String,
        target_id: Option<String>,
        message_type: String,
        payload: Vec<u8>,
        reliability: ReliabilityLevel,
        priority: MessagePriority,
        metadata: HashMap<String, String>,
    },
}

/// 槽位迁移类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SlotMigrationType {
    /// 开始迁移
    Start,
    /// 迁移数据
    Data,
    /// 完成迁移
    Complete,
    /// 取消迁移
    Cancel,
    /// 确认接收
    Acknowledge,
}

/// 故障类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum FailureType {
    /// 网络超时
    NetworkTimeout,
    /// 连接断开
    ConnectionLost,
    /// 节点无响应
    NodeUnresponsive,
    /// 数据不一致
    DataInconsistent,
    /// 服务不可用
    ServiceUnavailable,
}

/// 选举类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ElectionType {
    /// 主节点选举
    MasterElection,
    /// 配置变更选举
    ConfigElection,
    /// 故障恢复选举
    FailoverElection,
}

/// 负载信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LoadInfo {
    pub cpu_usage: f64,
    pub memory_usage: f64,
    pub active_connections: usize,
    pub qps: u64,
    pub slot_count: usize,
}

impl ProtocolMessageTrait for ClusterProtocolMessage {
    fn message_type(&self) -> &str {
        match self {
            Self::VersionNegotiation { .. } => "version_negotiation",
            Self::VersionNegotiationResponse { .. } => "version_negotiation_response",
            Self::StateSync { .. } => "state_sync",
            Self::StateSyncResponse { .. } => "state_sync_response",
            Self::SlotMigration { .. } => "slot_migration",
            Self::FailureDetection { .. } => "failure_detection",
            Self::Election { .. } => "election",
            Self::Heartbeat { .. } => "heartbeat",
            Self::Acknowledgment { .. } => "acknowledgment",
            Self::Custom { message_type, .. } => message_type,
        }
    }
    
    fn protocol_version(&self) -> u16 {
        CLUSTER_PROTOCOL_VERSION
    }
    
    fn sender_id(&self) -> &str {
        match self {
            Self::VersionNegotiation { sender_id, .. } |
            Self::VersionNegotiationResponse { sender_id, .. } |
            Self::StateSync { sender_id, .. } |
            Self::StateSyncResponse { sender_id, .. } |
            Self::SlotMigration { sender_id, .. } |
            Self::FailureDetection { sender_id, .. } |
            Self::Election { sender_id, .. } |
            Self::Heartbeat { sender_id, .. } |
            Self::Acknowledgment { sender_id, .. } |
            Self::Custom { sender_id, .. } => sender_id,
        }
    }
    
    fn target_id(&self) -> Option<&str> {
        match self {
            Self::VersionNegotiation { .. } => None,
            Self::VersionNegotiationResponse { target_id, .. } |
            Self::StateSyncResponse { target_id, .. } |
            Self::SlotMigration { target_id, .. } |
            Self::Acknowledgment { target_id, .. } => Some(target_id),
            Self::StateSync { target_id, .. } |
            Self::FailureDetection { target_id, .. } |
            Self::Election { target_id, .. } |
            Self::Heartbeat { target_id, .. } |
            Self::Custom { target_id, .. } => target_id.as_deref(),
        }
    }
    
    fn payload(&self) -> &[u8] {
        match self {
            Self::StateSync { payload, .. } |
            Self::StateSyncResponse { payload, .. } |
            Self::Election { payload, .. } |
            Self::Custom { payload, .. } => payload,
            Self::SlotMigration { data, .. } => data,
            Self::FailureDetection { evidence, .. } => evidence,
            _ => &[],
        }
    }
    
    fn set_payload(&mut self, new_payload: Vec<u8>) {
        match self {
            Self::StateSync { payload, .. } |
            Self::StateSyncResponse { payload, .. } |
            Self::Election { payload, .. } |
            Self::Custom { payload, .. } => *payload = new_payload,
            Self::SlotMigration { data, .. } => *data = new_payload,
            Self::FailureDetection { evidence, .. } => *evidence = new_payload,
            _ => {} // 其他类型不支持设置payload
        }
    }
    
    fn validate(&self) -> bool {
        // 基本验证：检查必要字段
        match self {
            Self::VersionNegotiation { sender_id, supported_versions, .. } => {
                !sender_id.is_empty() && !supported_versions.is_empty()
            }
            Self::VersionNegotiationResponse { sender_id, target_id, .. } => {
                !sender_id.is_empty() && !target_id.is_empty()
            }
            Self::StateSync { sender_id, .. } |
            Self::StateSyncResponse { sender_id, .. } |
            Self::SlotMigration { sender_id, .. } |
            Self::FailureDetection { sender_id, .. } |
            Self::Election { sender_id, .. } |
            Self::Heartbeat { sender_id, .. } |
            Self::Acknowledgment { sender_id, .. } |
            Self::Custom { sender_id, .. } => !sender_id.is_empty(),
        }
    }
    
    fn serialize(&self) -> ClusterResult<Vec<u8>> {
        bincode::serialize(self)
            .map_err(|e| ClusterError::Config(format!("协议消息序列化失败: {e}")))
    }
    
    fn deserialize(data: &[u8]) -> ClusterResult<Self> {
        bincode::deserialize(data)
            .map_err(|e| ClusterError::Config(format!("协议消息反序列化失败: {e}")))
    }
    
    fn reliability_level(&self) -> ReliabilityLevel {
        match self {
            Self::StateSync { reliability, .. } |
            Self::SlotMigration { reliability, .. } |
            Self::Custom { reliability, .. } => reliability.clone(),
            Self::VersionNegotiation { .. } |
            Self::VersionNegotiationResponse { .. } |
            Self::Acknowledgment { .. } => ReliabilityLevel::AtLeastOnce,
            Self::Election { .. } |
            Self::FailureDetection { .. } => ReliabilityLevel::BestEffort,
            Self::StateSyncResponse { .. } |
            Self::Heartbeat { .. } => ReliabilityLevel::BestEffort,
        }
    }
    
    fn priority(&self) -> MessagePriority {
        match self {
            Self::StateSync { priority, .. } |
            Self::SlotMigration { priority, .. } |
            Self::FailureDetection { priority, .. } |
            Self::Election { priority, .. } |
            Self::Heartbeat { priority, .. } |
            Self::Custom { priority, .. } => priority.clone(),
            Self::VersionNegotiation { .. } |
            Self::VersionNegotiationResponse { .. } => MessagePriority::High,
            Self::StateSyncResponse { .. } |
            Self::Acknowledgment { .. } => MessagePriority::Normal,
        }
    }
}

/// 集群协议处理器
pub struct ClusterProtocolProcessor {
    /// 本地节点ID
    local_node_id: String,
    
    /// 支持的协议版本
    supported_versions: Vec<u16>,
    
    /// 协议消息发送通道
    message_sender: mpsc::UnboundedSender<ClusterProtocolMessage>,
    
    /// 协议消息接收通道
    message_receiver: Option<mpsc::UnboundedReceiver<ClusterProtocolMessage>>,
    
    /// 广播通道
    broadcast_sender: broadcast::Sender<ClusterProtocolMessage>,
    
    /// 运行状态
    is_running: Arc<RwLock<bool>>,
    
    /// 协议统计信息
    stats: Arc<RwLock<ProtocolStats>>,
    
    /// 待确认消息 (用于可靠性保证)
    pending_acks: Arc<RwLock<HashMap<String, PendingMessage>>>,
}

/// 待确认的消息
#[derive(Debug, Clone)]
struct PendingMessage {
    message: ClusterProtocolMessage,
    send_time: Instant,
    retry_count: u32,
    max_retries: u32,
}

impl ClusterProtocolProcessor {
    /// 创建新的协议处理器
    pub fn new(local_node_id: String) -> ClusterResult<Self> {
        let (message_sender, message_receiver) = mpsc::unbounded_channel();
        let (broadcast_sender, _) = broadcast::channel(1000);
        
        let stats = ProtocolStats {
            messages_sent: 0,
            messages_received: 0,
            bytes_sent: 0,
            bytes_received: 0,
            protocol_errors: 0,
            version_mismatches: 0,
            reliability_retries: 0,
            average_latency: Duration::from_millis(0),
            peak_latency: Duration::from_millis(0),
            current_protocol_version: CLUSTER_PROTOCOL_VERSION,
            supported_versions: vec![MIN_SUPPORTED_VERSION, MAX_SUPPORTED_VERSION],
        };
        
        Ok(Self {
            local_node_id,
            supported_versions: vec![MIN_SUPPORTED_VERSION, MAX_SUPPORTED_VERSION],
            message_sender,
            message_receiver: Some(message_receiver),
            broadcast_sender,
            is_running: Arc::new(RwLock::new(false)),
            stats: Arc::new(RwLock::new(stats)),
            pending_acks: Arc::new(RwLock::new(HashMap::new())),
        })
    }
    
    /// 获取消息发送器
    pub fn message_sender(&self) -> mpsc::UnboundedSender<ClusterProtocolMessage> {
        self.message_sender.clone()
    }
    
    /// 订阅协议消息
    pub fn subscribe(&self) -> broadcast::Receiver<ClusterProtocolMessage> {
        self.broadcast_sender.subscribe()
    }
    
    /// 启动可靠性检查任务
    async fn start_reliability_checker(&self) -> ClusterResult<()> {
        let pending_acks = self.pending_acks.clone();
        let message_sender = self.message_sender.clone();
        let is_running = self.is_running.clone();
        let stats = self.stats.clone();
        
        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(1));
            
            while *is_running.read().await {
                interval.tick().await;
                
                let mut pending = pending_acks.write().await;
                let mut to_retry = Vec::new();
                let mut to_remove = Vec::new();
                
                for (msg_id, pending_msg) in pending.iter() {
                    let elapsed = pending_msg.send_time.elapsed();
                    
                    // 超时重试
                    if elapsed > Duration::from_secs(5) && pending_msg.retry_count < pending_msg.max_retries {
                        to_retry.push((msg_id.clone(), pending_msg.clone()));
                    }
                    // 超过最大重试次数，放弃
                    else if pending_msg.retry_count >= pending_msg.max_retries {
                        to_remove.push(msg_id.clone());
                        warn!("消息 {} 超过最大重试次数，放弃发送", msg_id);
                    }
                }
                
                // 执行重试
                for (msg_id, mut pending_msg) in to_retry {
                    pending_msg.retry_count += 1;
                    pending_msg.send_time = Instant::now();
                    
                    if let Err(e) = message_sender.send(pending_msg.message.clone()) {
                        error!("重试发送消息失败: {}", e);
                        to_remove.push(msg_id.clone());
                    } else {
                        pending.insert(msg_id, pending_msg);
                        
                        // 更新统计
                        let mut stats_guard = stats.write().await;
                        stats_guard.reliability_retries += 1;
                    }
                }
                
                // 移除失败的消息
                for msg_id in to_remove {
                    pending.remove(&msg_id);
                }
            }
        });
        
        Ok(())
    }
}

#[async_trait]
impl ClusterProtocol for ClusterProtocolProcessor {
    type ProtocolMessage = ClusterProtocolMessage;
    
    async fn start(&mut self) -> ClusterResult<()> {
        {
            let mut running = self.is_running.write().await;
            if *running {
                return Ok(());
            }
            *running = true;
        }
        
        info!("启动集群协议处理器...");
        
        // 启动可靠性检查任务
        self.start_reliability_checker().await?;
        
        // 启动消息处理任务
        let mut receiver = self.message_receiver.take()
            .ok_or_else(|| ClusterError::Config("消息接收器已被使用".to_string()))?;
        
        let broadcast_sender = self.broadcast_sender.clone();
        let is_running = self.is_running.clone();
        let stats = self.stats.clone();
        
        tokio::spawn(async move {
            while *is_running.read().await {
                match receiver.recv().await {
                    Some(message) => {
                        // 广播消息
                        if let Err(e) = broadcast_sender.send(message.clone()) {
                            warn!("广播协议消息失败: {}", e);
                        }
                        
                        // 更新统计
                        {
                            let mut stats_guard = stats.write().await;
                            stats_guard.messages_received += 1;
                            if let Ok(serialized) = ProtocolMessageTrait::serialize(&message) {
                                stats_guard.bytes_received += serialized.len() as u64;
                            }
                        }
                    }
                    None => {
                        debug!("协议消息通道已关闭");
                        break;
                    }
                }
            }
        });
        
        info!("集群协议处理器启动成功");
        Ok(())
    }
    
    async fn stop(&mut self) -> ClusterResult<()> {
        let mut running = self.is_running.write().await;
        *running = false;
        info!("集群协议处理器已停止");
        Ok(())
    }
    
    async fn send_protocol_message(&self, target: &str, message: Self::ProtocolMessage) -> ClusterResult<()> {
        // 检查可靠性要求
        let needs_ack = matches!(message.reliability_level(), ReliabilityLevel::AtLeastOnce | ReliabilityLevel::ExactlyOnce);
        
        if needs_ack {
            // 生成消息ID并加入待确认队列
            let msg_id = Uuid::new_v4().to_string();
            let pending_msg = PendingMessage {
                message: message.clone(),
                send_time: Instant::now(),
                retry_count: 0,
                max_retries: 3,
            };
            
            self.pending_acks.write().await.insert(msg_id, pending_msg);
        }
        
        // 发送消息
        self.message_sender.send(message.clone())
            .map_err(|e| ClusterError::Config(format!("发送协议消息失败: {e}")))?;
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.messages_sent += 1;
            if let Ok(serialized) = ProtocolMessageTrait::serialize(&message) {
                stats.bytes_sent += serialized.len() as u64;
            }
        }
        
        info!("发送协议消息到 {}: {}", target, message.message_type());
        Ok(())
    }
    
    async fn handle_protocol_message(&mut self, message: Self::ProtocolMessage) -> ClusterResult<()> {
        debug!("处理协议消息: {} 来自 {}", message.message_type(), message.sender_id());
        
        match &message {
            ClusterProtocolMessage::VersionNegotiation { sender_id, proposed_version, supported_versions, .. } => {
                // 处理版本协商
                let agreed_version = self.negotiate_version_with_peer(*proposed_version, supported_versions).await?;
                
                // 发送协商响应
                let response = ClusterProtocolMessage::VersionNegotiationResponse {
                    sender_id: self.local_node_id.clone(),
                    target_id: sender_id.clone(),
                    agreed_version,
                    timestamp: std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap().as_secs(),
                };
                
                self.send_protocol_message(sender_id, response).await?;
            }
            
            ClusterProtocolMessage::Acknowledgment { target_id, ack_message_id, success, .. } => {
                if target_id == &self.local_node_id {
                    // 移除待确认消息
                    self.pending_acks.write().await.remove(ack_message_id);
                    
                    if !success {
                        warn!("消息 {} 处理失败", ack_message_id);
                    }
                }
            }
            
            _ => {
                // 其他消息类型的具体处理逻辑
                debug!("处理其他类型协议消息: {}", message.message_type());
            }
        }
        
        Ok(())
    }
    
    async fn broadcast_protocol_message(&self, message: Self::ProtocolMessage) -> ClusterResult<()> {
        // 广播消息
        self.broadcast_sender.send(message.clone())
            .map_err(|e| ClusterError::Config(format!("广播协议消息失败: {e}")))?;
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.messages_sent += 1;
            if let Ok(serialized) = ProtocolMessageTrait::serialize(&message) {
                stats.bytes_sent += serialized.len() as u64;
            }
        }
        
        info!("广播协议消息: {}", message.message_type());
        Ok(())
    }
    
    async fn get_protocol_stats(&self) -> ProtocolStats {
        self.stats.read().await.clone()
    }
    
    fn is_version_compatible(&self, version: u16) -> bool {
        self.supported_versions.contains(&version)
    }
    
    async fn negotiate_version(&mut self, peer_version: u16) -> ClusterResult<u16> {
        self.negotiate_version_with_peer(peer_version, &[peer_version]).await
    }
}

impl ClusterProtocolProcessor {
    /// 与对等节点协商版本
    async fn negotiate_version_with_peer(&self, _peer_version: u16, peer_supported: &[u16]) -> ClusterResult<u16> {
        // 找到双方都支持的最高版本
        let mut common_versions: Vec<u16> = self.supported_versions
            .iter()
            .filter(|v| peer_supported.contains(v))
            .cloned()
            .collect();
        
        common_versions.sort_by(|a, b| b.cmp(a)); // 降序排列
        
        match common_versions.first() {
            Some(&version) => {
                info!("协商协议版本: {}", version);
                Ok(version)
            }
            None => {
                let error_msg = format!(
                    "无法协商协议版本: 本地支持 {:?}, 对方支持 {:?}",
                    self.supported_versions, peer_supported
                );
                error!("{}", error_msg);
                Err(ClusterError::Config(error_msg))
            }
        }
    }
}

